package com.example.cloudcommon.event.stream;

import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Service
public class EventStreamService {

    private final SseEmitter.SseEventBuilder eventBuilder = SseEmitter.event();

    // 存储活跃的SSE连接
    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();

    /**
     * 创建新的SSE连接
     */
    public SseEmitter createEmitter(String clientId) {
        SseEmitter emitter = new SseEmitter(3600000L); // 1小时超时

        // 设置超时和完成回调
        emitter.onTimeout(() -> {
            emitters.remove(clientId);
            log.info("SSE连接超时: {}", clientId);
        });

        emitter.onCompletion(() -> {
            emitters.remove(clientId);
            log.info("SSE连接完成: {}", clientId);
        });

        emitter.onError((ex) -> {
            emitters.remove(clientId);
            log.error("SSE连接错误: {}", clientId, ex);
        });

        emitters.put(clientId, emitter);

        // 发送连接成功事件
        sendEvent(clientId, EventStream.of("connected", "连接建立成功"));

        return emitter;
    }

    /**
     * 向特定客户端发送事件
     */
    public void sendEvent(String clientId, EventStream event) {
        SseEmitter emitter = emitters.get(clientId);
        if (emitter != null) {
            try {
                emitter.send(SseEmitter.event()
                        .id(event.getId())
                        .name(event.getEvent())
                        .data(event.getData())
                        .comment(event.getComment())
                        .reconnectTime(event.getRetry()));
            } catch (Exception e) {
                log.error("发送事件失败: {}", clientId, e);
                emitters.remove(clientId);
            }
        }
    }

    /**
     * 广播事件给所有客户端
     */
    public void broadcast(EventStream event) {
        emitters.forEach((clientId, emitter) -> {
            sendEvent(clientId, event);
        });
    }

    /**
     * 关闭指定客户端连接
     */
    public void closeEmitter(String clientId) {
        SseEmitter emitter = emitters.get(clientId);
        if (emitter != null) {
            emitter.complete();
            emitters.remove(clientId);
        }
    }

    /**
     * 获取活跃连接数
     */
    public int getActiveConnections() {
        return emitters.size();
    }
}